#include "stdafx.h"
#include "log/asyncchannel.hpp"
#include "log/formatter.hpp"
#include "log/logregistry.hpp"
#include "stdlib/bugcheck.hpp"
#include "stdlib/exception.hpp"
using namespace JHCPP::stdlib;
using namespace JHCPP::stdlib::jhException;
#include "utils/autoptr.hpp"
using JHCPP::utils::TAutoPtr;


namespace JHCPP
{
	namespace log
	{
		class CMessageNotification: public CNotification
		{
		public:
			CMessageNotification(const CMessage& msg) : m_msg(msg)
			{
			}

			~CMessageNotification()
			{
			}

			const CMessage& message() const
			{
				return m_msg;
			}

		private:
			CMessage m_msg;
		};


		CAsyncChannel::CAsyncChannel(CChannel* pChannel, CThread::Priority prio)
		: m_pChannel(pChannel), m_thread("AsyncChannel")
		{
			if (m_pChannel) 
				m_pChannel->add();

			m_thread.setPriority(prio);
		}

		CAsyncChannel::~CAsyncChannel()
		{
			try
			{
				close();
				if (m_pChannel) 
					m_pChannel->release();
			}
			catch (...)
			{
				jh_unexpected();
			}
		}

		void CAsyncChannel::setChannel(CChannel* pChannel)
		{
			CMutex::ScopedLock lock(m_channelMutex);

			if (m_pChannel) m_pChannel->release();
			m_pChannel = pChannel;
			if (m_pChannel) m_pChannel->add();
		}

		CChannel* CAsyncChannel::getChannel() const
		{
			return m_pChannel;
		}

		void CAsyncChannel::open()
		{
			CMutex::ScopedLock lock(m_threadMutex);

			if (!m_thread.isRunning())
				m_thread.start(*this);
		}

		void CAsyncChannel::close()
		{
			if (m_thread.isRunning())
			{
				while (!m_queue.empty()) CThread::sleep(100);

				do 
				{
					m_queue.wakeUpAll(); 
				}
				while (!m_thread.tryJoin(100));
			}
		}

		void CAsyncChannel::log(const CMessage& msg)
		{
			open();

			m_queue.enqueueNotification(new CMessageNotification(msg));
		}

		void CAsyncChannel::setProperty(const std::string& name, const std::string& value)
		{
			if (name == "channel")
				setChannel(CLogRegistry::defaultRegistry().channelForName(value));
			else if (name == "priority")
				setPriority(value);
			else
				CChannel::setProperty(name, value);
		}

		void CAsyncChannel::run()
		{
			TAutoPtr<CNotification> nf = m_queue.waitDequeueNotification();
			while (nf)
			{
				CMessageNotification* pNf = dynamic_cast<CMessageNotification*>(nf.get());
				{
					CMutex::ScopedLock lock(m_channelMutex);

					if (pNf && m_pChannel) m_pChannel->log(pNf->message());
				}
				nf = m_queue.waitDequeueNotification();
			}
		}

		void CAsyncChannel::setPriority(const std::string& value)
		{
			CThread::Priority prio = CThread::PRIO_NORMAL;

			if (value == "lowest")
				prio = CThread::PRIO_LOWEST;
			else if (value == "low")
				prio = CThread::PRIO_LOW;
			else if (value == "normal")
				prio = CThread::PRIO_NORMAL;
			else if (value == "high")
				prio = CThread::PRIO_HIGH;
			else if (value == "highest")
				prio = CThread::PRIO_HIGHEST;
			else
				throw InvalidArgumentException("thread priority", value);

			m_thread.setPriority(prio);
		}
	}//end of namespace log
}//end of namespace JHCPP
